package com.niit.handler

import com.google.gson.Gson
import com.niit.bean.{AdClickData, Answer}
import com.niit.utils.{MyKafkaUtil, SparkUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.streaming.dstream.{DStream, InputDStream}

/*
 消费Kafka中的数据，并将数据进行格式转换 格式处理
 */
object DataHandler {

  private val scc = SparkUtil.takeSSC()

  //处理Kafka中的数据，将 字符串中的数据 封装到 样例类当中
  def kafkaDataHandler(groupId:String,topic:String):DStream[AdClickData]={
    val kfDataDS: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(groupId, topic, scc)
    val adData: DStream[AdClickData] = kfDataDS.map(data => { //data 就是读取Kafka中的每一条数据
      val adClickData = data.value() //1714960532145 华东 深圳 6 2
      val datas = adClickData.split(" ")
      AdClickData(datas(0), datas(1), datas(2), datas(3), datas(4))
    })
    adData
  }

  /* 读取Kafka中的数据 ，并对类似下列格式的数据进行解析，解析后的数据封装到Answer
  {"student_id":"学生ID_39",
  "textbook_id":"教材ID_1",
  "grade_id":"年级ID_4",
  "subject_id":"科目ID_2_语文",
  "chapter_id":"章节ID_chapter_3",
  "question_id":"题目ID_660",
  "score":4,
  "answer_time":"2024-05-27 09:51:38",
  "ts":"May 27, 2024 9:51:38 AM"}
   */

  def kafkaAnswerDataHandler(groupId:String,topic:String):DStream[Answer]={
    //获得一批Kafka中的数据
    val kfDataDS: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(groupId, topic, scc)
    val resData: DStream[Answer] = kfDataDS.map(kfData => { //kfData 每一条数据
      val value: String = kfData.value() //json的格式字符串
    val gson = new Gson() // 是一个JSON数据格式的转换工具 可以将字符串、类与JSON格式进行相互转换
    //将json格式字符串 转成 Answer样例类
    /*
     转换的注意事项：
            1.JSON的中Key 一定要在样例类当前存在，如果不存在，转换后会丢失数据
            2.JSON的中Key的名字一定要和样例类属性名一致，大小写、空格、格式完全一样
                                         特殊格式  JSON: student_name  样例类 studentName

     */
    val answer: Answer = gson.fromJson(value, classOf[Answer])
      answer
    })
    resData


  }


  //让ssc开启采集和等待采集
  def startAndAwait(): Unit ={
    scc.start();
    scc.awaitTermination()
  }

}
